package netbolt

import (
	"github.com/boltdb/bolt"
	"github.com/samuel/go-zookeeper/zk"
	"os"
	"path/filepath"
	"strconv"
	"sync"
	"time"
)

//var SYSDB *NetBoltDB = &NetBoltDB{"sys", time.Now(), NetBoltConfig.Netbolt.DBPath + "/sys.db"}
//var (
//	NOT_FOUND_DB_ERROR = errors.New("NOT_FOUND_DB")
//)
var (
	sysDB              *NetBoltDB
	DataSerializer, _  = NewSerializer(GOB)
	dbLock             = &sync.RWMutex{}
	changedDBNames     = make(map[string]uint8)
	changedDBNamesLock = &sync.Mutex{}
	//cursorLock         = &sync.RWMutex{}
	cursors = sync.Map{}
)

const (
	FIRST int32 = 0x0
	NEXT  int32 = 0x1
	PREV  int32 = 0x2
	LAST  int32 = 0x3
	SEEK  int32 = 0x4
)

//数据库信息
type NetBoltDB struct {
	Name     string
	FilePath string
	Created  time.Time
	Compress string
	Slots    [SLOT_COUNT]bool
}

//数据库实例
type NetBoltDBInstance struct {
	NetBoltDB
	DBInstance *bolt.DB
}

func ExistDB(name string) (bool, error) {
	ret, _, err := ZkConn.Exists(ZK_DB_PATH + "/" + name)
	return ret, err
}

func timingUpdateDB() {
	changedDBNamesLock.Lock()
	defer changedDBNamesLock.Unlock()

	if len(changedDBNames) > 0 {
		dbLock.RLock()
		defer dbLock.RUnlock()
		for name := range changedDBNames {
			if dbv, ok := Databases.Load(name); ok {
				db := dbv.(*NetBoltDBInstance)
				db.Update()
				if DEBUG {
					DEBUGLog.Printf("sync %s db mete data,db size:%d,db slots:%d ", db.Name, db.Size(), db.GetSlotSize())
				}

			}

		}

		changedDBNames = make(map[string]uint8)
	}

}

func ListDB() ([]NetBoltDB, error) {

	names, _, err := ZkConn.Children(ZK_DB_PATH)
	var dbs = make([]NetBoltDB, len(names))

	for i, n := range names {
		db, err := LoadDB(n)
		if err != nil {
			return nil, err
		}
		dbs[i] = *db
	}
	return dbs, err
}

func CreateDB(name string) (*NetBoltDB, error) {

	dbLock.Lock()
	defer dbLock.Unlock()
	if exist, _ := ExistDB(name); exist {
		return nil, DB_EXIST_ERROR
	}

	dbFileName := UUID() + ".db"
	d := &NetBoltDB{Name: name, FilePath: name + "/" + dbFileName, Created: time.Now(), Compress: "gzip"}
	created, err := d.create()
	if err != nil {
		return d, err
	}

	if !created {
		return nil, CREATE_DB_ERROR
	}
	return d, nil

}

func CreateDBAndOpen(name string) (*NetBoltDBInstance, error) {
	d, err := CreateDB(name)
	if err != nil {
		return nil, err
	}

	return d.open()

}

func DropDB(name string, deleteFile bool) error {
	if wddb, ok := Databases.Load(name); ok {
		dbLock.Lock()
		defer dbLock.Unlock()

		instance := wddb.(*NetBoltDBInstance)
		instance.Close()

		err := ZkConn.Delete(ZK_DB_PATH+"/"+name, -1)
		if err != nil {
			return err
		}

		if deleteFile {
			err = os.Remove(instance.FilePath)
			if err != nil {
				return err
			}
		}

		Databases.Delete(name)

		Logger.Printf("Drop db : %s.", name)
		return nil

	} else {
		return DB_NOT_FOUND
	}

}

func OpenDB(name string) (*NetBoltDBInstance, error) {
	dbInfo, err := LoadDB(name)
	if err != nil {
		return nil, err
	}

	return dbInfo.open()
}

func (d *NetBoltDBInstance) Close() error {
	if d.DBInstance != nil {
		return d.DBInstance.Close()
	} else {
		return nil
	}
}

//func NewDB(name string, filePath string) {
//	if len(name)>256{
//		log.Println("db name size > 256!")
//	}
//	if(filePath)
//
//}
//拉去DB基础信息
func LoadDB(name string) (*NetBoltDB, error) {
	exist, err := ExistDB(name)
	if err != nil {
		return nil, DB_NOT_FOUND
	}

	if exist {
		b, _, err := ZkConn.Get(ZK_DB_PATH + "/" + name)
		if err != nil {
			return nil, err
		}
		db := &NetBoltDB{}
		//buf := bytes.NewBuffer(b)
		//binary.Read(buf, binary.BigEndian, db)

		err = DataSerializer.FromBytes(b, db)
		if err != nil {
			return nil, err
		}
		return db, err
	} else {
		return nil, nil
	}
}

//如果数据库存在或者数据库创建出错（同时带回错误）则返回false
func (db *NetBoltDB) create() (bool, error) {
	exist, err := ExistDB(db.Name)
	if err != nil {
		return false, err
	}
	db.Created = time.Now()
	if !exist {

		exist, _ = FileExists(db.FilePath)

		//dbi, err := bolt.Open(db.FilePath, 0600, nil)
		//defer dbi.Close()

		buf, err := DataSerializer.ToBytes(*db)

		if err != nil {
			ErrLog.Panic(err)
		}
		if _, err = ZKCreateIfNotExist(ZK_DB_PATH+"/"+db.Name, buf, 0, zk.WorldACL(zk.PermAll)); err != nil && err != zk.ErrNodeExists {
			return false, err
		} else {
			return true, nil
		}
	} else {
		return false, nil
	}
}

//单个事务以及从它们创建的所有对象（例如bucket，key）不是线程安全的。 要处理多个goroutine 中的数据，您必须为每个 goroutine 启动一个事务，或使用锁来确保一次只有一个 goroutine 访问事务。 从 DB 创建事务是线程安全的。
//只读事务和读写事务不应该相互依赖，一般不应该在同一个例程中同时打开。 这可能会导致死锁，因为读写事务需要定期重新映射数据文件，但只有在只读事务处于打开状态时才能这样做。
func (db *NetBoltDB) open() (*NetBoltDBInstance, error) {
	dbLock.Lock()
	defer dbLock.Unlock()
	if ret, ok := Databases.Load(db.Name); ok {
		return ret.(*NetBoltDBInstance), nil
	} else {
		absdir, err := filepath.Abs(NetBoltConfig.Netbolt.DBPath + "/" + db.FilePath)
		if err != nil {
			return nil, err
		}
		dir := filepath.Dir(absdir)

		if exist, _ := FileExists(dir); !exist {
			os.MkdirAll(dir, 0777)
		}

		dbInstance, err := bolt.Open(NetBoltConfig.Netbolt.DBPath+"/"+db.FilePath, 0777, nil)

		var curSize int = 0
		dbInstance.Update(func(tx *bolt.Tx) error {
			b, err := tx.CreateBucketIfNotExists(indexBucketName)
			curSize = b.Stats().KeyN

			return err
		})

		var ret *NetBoltDBInstance
		if err == nil {
			ret = &NetBoltDBInstance{*db, dbInstance}

			if _, ok := Databases.Load(db.Name); !ok {
				Databases.Store(db.Name, ret)
			}
			Logger.Printf("Database %s load complete,row size:%d.\n", ret.NetBoltDB.Name, curSize)
		}

		return ret, err
	}

}

func (db *NetBoltDB) ExistSlot(num uint16) bool {
	//index := sort.SearchInts(db.Slots, num)
	//return index < len(db.Slots)

	return db.Slots[num]
}

func (db *NetBoltDB) Update() error {
	buf, err := DataSerializer.ToBytes(*db)
	if err != nil {
		return err
	}

	_, err = ZkConn.Set(ZK_DB_PATH+"/"+db.Name, buf, -1)

	return err

}

func (db *NetBoltDB) changed() {
	changedDBNamesLock.Lock()
	defer changedDBNamesLock.Unlock()

	_, ok := changedDBNames[db.Name]

	if !ok {
		changedDBNames[db.Name] = 0
	}

}

func (db *NetBoltDB) GetSlotSize() int16 {
	var count int16
	for _, b := range db.Slots {
		if b {

			count++
		}
	}

	return count
}

func (db *NetBoltDB) AddSlot(num uint16) (bool, error) {

	if !db.ExistSlot(num) {

		db.Slots[num] = true
		//err := db.update()
		//if err != nil {
		//	db.Slots[num] = false
		//	return false, err
		//}
		db.changed()
		return true, nil
	} else {
		return false, nil
	}

}

func getSlotName(key []byte) []byte {
	hashcode := CRC_CCITT(key)
	slotNum := hashcode % SLOT_COUNT
	slotBuf := strconv.Itoa(int(slotNum))
	return []byte(slotBuf)
}

func (db *NetBoltDBInstance) Get(key []byte) ([]byte, error) {
	hashcode := CRC_CCITT(key)
	slotNum := hashcode % SLOT_COUNT

	if !db.ExistSlot(slotNum) {
		return nil, nil
	}

	slotBuf := strconv.Itoa(int(slotNum))

	dbInstance := db.DBInstance
	tx, err := dbInstance.Begin(false)

	if err != nil {

		return nil, err
	}

	defer func() {
		if tx != nil {
			tx.Rollback()
		}
	}()
	slot := tx.Bucket([]byte(slotBuf))

	value := slot.Get(key)

	if value != nil {
		switch db.Compress {
		case "gzip":
			cb, err := GZIPUnCompress(value)
			if err != nil {
				return nil, err
			}

			return cb, nil

		default:
			return value, nil
		}
	} else {
		return nil, nil
	}

}

func (db *NetBoltDBInstance) Put(key []byte, value []byte) error {

	dbInstance := db.DBInstance
	hashcode := CRC_CCITT(key)
	slotNum := hashcode % SLOT_COUNT
	slotBuf := []byte(strconv.Itoa(int(slotNum)))
	tx, err := dbInstance.Begin(true)

	if err != nil {

		return err
	}
	defer func() {
		if tx != nil {
			tx.Rollback()
		}
	}()

	indexSlot, err := tx.CreateBucketIfNotExists(indexBucketName)

	if err != nil {
		return err
	}

	indexSlot.Put(key, slotBuf)

	slot, err := tx.CreateBucketIfNotExists(slotBuf)

	if err != nil {

		return err
	}

	createdSlot, err := db.AddSlot(slotNum)
	if err != nil {
		return err
	}
	if DEBUG && createdSlot {
		DEBUGLog.Printf("create new slot:%d\n", slotNum)
	}

	switch db.Compress {
	case "gzip":
		cb, err := GZIPCompress(value)
		if err != nil {
			return err
		}
		err = slot.Put(key, cb)
	default:
		err = slot.Put(key, value)
	}
	if err != nil {

		return err
	}

	if err := tx.Commit(); err != nil {
		return err
	}

	if createdSlot {
		db.changed()
	}
	return nil
}

//func (db *NetBoltDBInstance) NextSequence() uint64{
//
//}
func (db *NetBoltDBInstance) BatchPut(entity []*Entity) error {
	dbInstance := db.DBInstance

	createSlots := make([]uint16, 0)

	err := dbInstance.Batch(func(tx *bolt.Tx) error {

		for _, e := range entity {
			hashcode := CRC_CCITT(e.Key)
			slotNum := hashcode % SLOT_COUNT

			slotBuf := strconv.Itoa(int(slotNum))
			slotName := []byte(slotBuf)

			indexSlot, err := tx.CreateBucketIfNotExists(indexBucketName)
			if err != nil {
				return err
			}

			indexSlot.Put(e.Key, slotName)

			b, err := tx.CreateBucketIfNotExists(slotName)
			if err != nil {
				return err
			}

			createSlots = append(createSlots, slotNum)

			switch db.Compress {
			case "gzip":
				cb, err := GZIPCompress(e.Value)
				if err != nil {
					return err
				}
				err = b.Put(e.Key, cb)
			default:
				err = b.Put(e.Key, e.Value)
			}

			if err != nil {
				return err
			}

		}

		return nil
	})

	if err == nil {

		newSolt := false
		for _, s := range createSlots {
			createdSlot, _ := db.AddSlot(s)
			if DEBUG && createdSlot {
				newSolt = true
				DEBUGLog.Printf("create new slot:%d\n", s)
			}
		}

		if newSolt {
			db.changed()
		}

	}

	return err
}

type NetBoltCursor struct {
	Cursor
	dbi  *bolt.DB
	tx   *bolt.Tx
	b    *bolt.Bucket
	c    *bolt.Cursor
	init bool
	//step  int32
	//index int64
}

//func (db *NetBoltDBInstance) CreateCursor(token string) (*Cursor, error) {
//	id := UUID()
//	cursor := &Cursor{Id: id, Token: token}
//
//	dbi := db.DBInstance
//	tx, err := dbi.Begin(false)
//	if err != nil {
//		if tx != nil {
//			tx.Rollback()
//		}
//		return nil, err
//	}
//	b := tx.Bucket(indexBucketName)
//	if err != nil {
//		if tx != nil {
//			tx.Rollback()
//		}
//		return nil, err
//	}
//	c := b.Cursor()
//
//	nbc := &NetBoltCursor{*cursor, dbi, tx, b, c, false}
//	cursors.Store(id, nbc)
//	return cursor, nil
//}

func (db *NetBoltDBInstance) Size() int {
	dbi := db.DBInstance
	var size int = 0
	dbi.View(func(tx *bolt.Tx) error {
		slot := tx.Bucket(indexBucketName)
		size = slot.Stats().KeyN
		return nil
	})
	return size
}

func (db *NetBoltDBInstance) Delete(key []byte) error {
	dbi := db.DBInstance
	return dbi.Update(func(tx *bolt.Tx) error {
		hashcode := CRC_CCITT(key)
		slotNum := hashcode % SLOT_COUNT

		slotBuf := strconv.Itoa(int(slotNum))
		slotName := []byte(slotBuf)

		b := tx.Bucket(slotName)

		if b == nil {
			return nil
		}
		err := b.Delete(key)

		if err != nil {
			return err
		}
		index := tx.Bucket(indexBucketName)
		err = index.Delete(key)
		if err != nil {
			return err
		}
		return nil
	})
}

func (db *NetBoltDBInstance) BatchDelete(keys [][]byte) error {
	dbi := db.DBInstance

	return dbi.Update(func(tx *bolt.Tx) error {

		for _, key := range keys {
			hashcode := CRC_CCITT(key)
			slotNum := hashcode % SLOT_COUNT

			slotBuf := strconv.Itoa(int(slotNum))
			slotName := []byte(slotBuf)

			b := tx.Bucket(slotName)

			if b == nil {
				return nil
			}
			err := b.Delete(key)

			if err != nil {
				return err
			}
			index := tx.Bucket(indexBucketName)
			err = index.Delete(key)
			if err != nil {
				return err
			}

		}
		return nil
	})
}
func (db *NetBoltDBInstance) Stream(op *StreamOp, handler func(entry *Entry) (*StreamOp, error)) error {

	dbi := db.DBInstance

	return dbi.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(indexBucketName)

		c := b.Cursor()

		//init := false

		var err error
		for op != nil && err == nil { //需要优化 改成不用服务端来处理First()
			var ks [][]byte
			//DEBUGLog.Printf("%v\n", op)
			if op.Size == 0 {
				op.Size = 1
			}
			//if !init {
			//	switch op.Flg {
			//	case NEXT:
			//		init = true
			//
			//		k, _ := c.First()
			//		var null bool
			//		if k == nil {
			//			null = true
			//		} else {
			//			ks = append(ks, k)
			//		}
			//
			//		if !null {
			//			var i int32 = 1
			//			for i = 1; i < op.Size; i++ {
			//				k, _ := c.Next()
			//				if k == nil {
			//					null = true
			//					break
			//				}
			//
			//				ks = append(ks, k)
			//			}
			//		}
			//		op, err = handler(&Entry{Key: ks, Finish: null}) //发送数据,设置下一步操作
			//	case PREV:
			//		init = true
			//		c.First()
			//
			//		op, err = handler(&Entry{Key: ks, Finish: true})
			//	case LAST:
			//		init = true
			//
			//		k, _ := c.Last()
			//		var null bool
			//
			//		if k == nil {
			//			null = true
			//		} else {
			//			ks = append(ks, k)
			//		}
			//
			//		if !null {
			//			var i int32 = 1
			//			for i = 1; i < op.Size; i++ {
			//				k, _ := c.Prev()
			//				if k == nil {
			//					null = true
			//					break
			//				}
			//
			//				ks = append(ks, k)
			//			}
			//		}
			//		op, err = handler(&Entry{Key: ks, Finish: null})
			//	case SEEK:
			//		init = true
			//		k, _ := c.Seek(op.SeekKey)
			//		op, err = handler(&Entry{Key: [][]byte{k}, Finish: k == nil})
			//	}
			//
			//} else {
			switch op.Flg {
			case NEXT:

				var i int32
				var null bool
				for i = 0; i < op.Size; i++ {
					k, _ := c.Next()
					if k == nil {
						null = true
						break
					}

					ks = append(ks, k)
				}

				op, err = handler(&Entry{Key: ks, Finish: null})
			case FIRST:

				k, _ := c.First()
				var null bool
				if k == nil {
					null = true
				} else {
					ks = append(ks, k)
				}

				if !null {
					var i int32 = 1
					for i = 1; i < op.Size; i++ {
						k, _ := c.Next()
						if k == nil {
							null = true
							break
						}

						ks = append(ks, k)
					}
				}
				op, err = handler(&Entry{Key: ks, Finish: null})
			case PREV:

				var i int32 = 0
				var null bool
				for i = 0; i < op.Size; i++ {
					k, _ := c.Prev()
					if k == nil {
						null = true
						break
					}

					ks = append(ks, k)
				}

				op, err = handler(&Entry{Key: ks, Finish: null})
			case LAST:

				k, _ := c.Last()
				var null bool

				if k != nil {
					null = true
				} else {
					ks = append(ks, k)
				}

				if !null {
					var i int32 = 1
					for i = 1; i < op.Size; i++ {
						k, _ := c.Prev()
						if k == nil {
							null = true
							break
						}

						ks = append(ks, k)
					}
				}
				op, err = handler(&Entry{Key: ks, Finish: null})
			case SEEK:

				k, _ := c.Seek(op.SeekKey)
				op, err = handler(&Entry{Key: [][]byte{k}, Finish: k == nil})
			default:
				op, err = handler(&Entry{Key: [][]byte{}, Finish: true})
			}
		}

		//}

		return nil
	})

}

//func (db *NetBoltDBInstance) Next(c *Cursor) (*Entry, error) {
//
//	nc, has := cursors.Load(c.Id)
//	if !has {
//		return nil, NOT_OPEN_CURSOR
//	}
//
//	cu := nc.(*NetBoltCursor)
//	//TODO:取出实际值
//
//	if !cu.init {
//		k, v := cu.c.First()
//		cu.init = true
//		return &Entry{Key: k, Value: v}, nil
//	} else {
//		k, v := cu.c.Next()
//
//		return &Entry{Key: k, Value: v}, nil
//	}
//
//}
